package com.yujiahao.bigdata.wordcount

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

object WordCount {
  def main(args: Array[String]): Unit = {

    //TODO 1、获得Spark环境的连
    val conf = new SparkConf().setAppName("WordCount").setMaster("local[*]")
    val sc = new SparkContext(conf)

    //TODO 3、业务逻辑 --用十种方法求WordCount

    //这里是直接给定一个字符串list集合，当然也可以从文件中读取
    val rdd: RDD[String] = sc.makeRDD(
      List(
        "hello", "spark", "hello", "scala"
      )
    )

    //设置缓存一下，优化一下
    rdd.cache()

    //TODO 方式一：groupBy
    println("----------------------方式一：groupBy-------------------------")
    //现将原始集合格式进行转换，用Key进行分组，value值会是专门存放数量1的集合，拿到长度即可
    rdd.map((_,1)).groupBy(_._1).mapValues(_.size).collect().foreach(println)

    //TODO 方式二：reduceBykey
    println("----------------------方式二：reduceBykey -------------------------")
    //现将原始集合格式进行转换，直接使用reduceByKey就行，相同key的values按照定义规则聚合即可
    rdd.map((_,1)).reduceByKey(_+_).collect().foreach(println)

    //TODO 方式三：groupByKey
    println("----------------------方式三：groupByKey-------------------------")
    //现将原始集合格式进行转换，用Key进行分组，value值然后聚合遍历即可
    rdd.map((_,1)).groupByKey().mapValues(_.size).collect().foreach(println)

    //TODO 方式四：aggregateByKey
    println("----------------------方式四：aggregateByKey-------------------------")
    //现将原始集合格式进行转换，使用aggregateByKey，传入初始值，在进行自定义分区内和分区间规则
    rdd.map((_,1)).aggregateByKey(0)(_+_,_+_).collect().foreach(println)

    //TODO 方式五：foldByKey
    println("----------------------方式五：foldByKey-------------------------")
    //现将原始集合格式进行转换，特殊的aggregateByKey，分区内和分区间规则相同使用fold
    rdd.map((_,1)).foldByKey(0)(_+_).collect().foreach(println)

    //TODO 方式六：combineByKey
    println("----------------------方式六：combineByKey-------------------------")
    //现将原始集合格式进行转换，使用combinerByKey改变第一个相同k的v的格式，接着进行分区内分区间的转化
    rdd.map((_,1)).combineByKey(
      v =>v,
      (x:Int,v)=>{
      x+v
    },
      (t1:Int,t2:Int)=>{
        t1+t2
      }
    ).collect().foreach(println)

    //TODO 方式七：countByKey
    println("----------------------方式七：countByKey-------------------------")

    rdd.map((_, 1)).countByKey().map(word => {
      (word._1, word._2)
    }).foreach(println)


    //TODO 方式八：countByValue
    println("----------------------方式八：countByValue-------------------------")
    val map: collection.Map[(String, Int), Long] = rdd.map((_, 1)).countByValue()
    //上面转换为是Map集合，所以这里就转变一下格式
    map.map(num=> println({
      (num._1._1, num._2)
    }))


    //TODO 方式九：aggregate
    println("----------------------方式九：aggregate-------------------------")
    rdd.map((_, 1)).aggregate(mutable.Map[String, Int]())(
      (map, kv) => {
        val word: String = kv._1
        val num: Int = kv._2
        map(word) = map.getOrElse(word, 0) + num
        map
      },
      (map1, map2) => {
        map1.foldLeft(map2) {
          case (map, kv) => {
            map(kv._1) = map.getOrElse(kv._1, 0) + kv._2
            map
          }
        }
      }
    ).foreach(println)

    //TODO 方式十：fold
    println("----------------------方式十：fold-------------------------")

    rdd.map((_, 1)).map(mutable.Map(_)).fold(mutable.Map[String, Int]())(
      (map1,map2) => {
        map1.foldLeft(map2){
          case (map, kv) => {
            map(kv._1) = map.getOrElse(kv._1, 0) + kv._2
            map
          }
        }
      }
    ).foreach(println)
    //TODO 方式十一：reduce
    println("----------------------方式十一：reduce-------------------------")
    //这里的实现思路是reduce是一个聚合的行动算子，将数据进行两两聚合并直接返回结果，(String,String)=> String
    //要想实现WordCount也就是说数据要进行两两聚合最终的结果应该是一个键值对（Word，count），
    // 根据reduce的参数以及返回值类型可以得出将原始数据作为Word并转为可变的Map集合在聚合，使用map转换格式不就好了嘛
    //那么就说在reduce阶段是两个集合之间的合并，可以使用模式匹配
    val result: mutable.Map[String, Int] = rdd.map(
      word => {
        mutable.Map[String, Int]((word, 1))
      }
    ).reduce {
      (map1, map2) => {
        //这里进行两个map的合并
        map1.foreach {
          case (word, cnt) => {
            val oldCnt: Int = map2.getOrElse(word, 0)
            map2.update(word, oldCnt + cnt)
          }
        }
        map2
      }
    }

    println(result)
    //TODO 2、关闭Spark环境的连接
   sc.stop()
  }
}